package com.rong.distributed.transaction.mq;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import com.rong.distributed.transaction.mq.bean.Message;

public class MessageQueue {

	private static final MessageQueue MESSAGE_QUEUE = new MessageQueue();

	/**
	 * mq队列，key：topic value:队列
	 */
	private Map<String, BlockingQueue<Message>> QUEUE_CACHE = new ConcurrentHashMap<String, BlockingQueue<Message>>();

	private MessageQueue() {

	}

	public static MessageQueue getInstance() {
		return MESSAGE_QUEUE;
	}

	public void putMessage(Message message) {
		String topic = message.getTopic();
		BlockingQueue<Message> queue = QUEUE_CACHE.get(topic);
		if (queue != null) {
			queue.offer(message);
			return;
		}

		synchronized (this) {
			queue = QUEUE_CACHE.get(topic);
			if (queue != null) {
				queue.offer(message);
				return;
			}

			queue = new LinkedBlockingQueue<Message>();
			queue.offer(message);
			QUEUE_CACHE.put(message.getTopic(), queue);
		}
	}

	public Message pollMessage(String topic) throws Exception {
		BlockingQueue<Message> queue = QUEUE_CACHE.get(topic);
		if (queue != null) {
			return queue.take();
		}

		synchronized (this) {
			queue = QUEUE_CACHE.get(topic);
			if (queue != null) {
				return queue.take();
			}

			queue = new LinkedBlockingQueue<Message>();
			QUEUE_CACHE.put(topic, queue);

			return queue.take();
		}
	}
}
